/* Copyright  2009-2010
   @Author
   Richard Changde Yin            e-mail yinchangde@hotmail.com

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/*
* @filename :  databaseManagerImpl.cc
*/
#include<server_includes.h>
#include<database.h>
#include<catalogtables.h>
#include<databaseManager.h>
#include<databaseManagerImpl.h>
#include<os.h>
#include<table.h>

#include<transaction.h>
#include<catalogtables.h>
#include<index.h>


#include<lock.h>
#include<debug.h>
#include<read-config.h>
#include<process.h>

// for std
#include <string.h>

#include <m_string.h>

#include <gettext.h>
#include <error.h>
#include <errmsg_print.h>

#include <session.h>
#include <current_session.h>
#include <sql_error.h>

using namespace BackEnd ;

Database* DatabaseManagerImpl::g_systemDatabase = 0;
Database* DatabaseManagerImpl::g_database       = 0;


#define   FILL_IN_INDX_INFO(IDXINFOPTR)                                \
                        /* fill in the tbl name */                     \
                      strcpy((IDXINFOPTR)->tableName,name );           \
                      strcpy((IDXINFOPTR)->idxName,_idxname.c_str());  \
                      (IDXINFOPTR)->length   =  key_part->length;     \
                      (IDXINFOPTR)->ha_type =   key_part->type ;      \
                      if (key_part->type == HA_KEYTYPE_VARBINARY1 ||  \
                          key_part->type == HA_KEYTYPE_VARTEXT1)      \
                      {                                               \
                         IDXINFOPTR->bit_start = 1 ;                  \
                         (IDXINFOPTR)->length  +=2 ;                  \
                      }                                               \
                      if (key_part->type == HA_KEYTYPE_VARBINARY2 ||  \
                          key_part->type == HA_KEYTYPE_VARTEXT2)      \
                      {                                               \
                         IDXINFOPTR->bit_start = 2 ;                  \
                         (IDXINFOPTR)->length  +=2 ;                  \
                      }                                               \
                       if (field->flags & (ENUM_FLAG | SET_FLAG))      \
                       /*     ctype_bin.cc*/                           \
                         (IDXINFOPTR)->charset= &my_charset_bin;       \
                       else                                            \
                         (IDXINFOPTR)->charset= field->charset();      \
                       if (field->null_ptr)                            \
                       {                                               \
                           (IDXINFOPTR)->null_bit= field->null_bit;          \
                           (IDXINFOPTR)->null_pos=  \
              (uint) (field->null_ptr - (unsigned char*)tbl->record[0]);      \
                        }                                                \
                        else                                             \
                        {                                                \
                          (IDXINFOPTR)->null_bit= 0;                     \
                          (IDXINFOPTR)->null_pos= 0;                     \
                        }                                                \
                        /* fldoffset*/                                   \
    (IDXINFOPTR)->start=(IDXINFOPTR)->fldOffset=  key_part->offset  ;    \
                        (IDXINFOPTR)->fieldnr=    key_part->fieldnr ;    \
                        (IDXINFOPTR)->type = field->type();              \
                         if ( primary_key_no ==(int)i )                  \
                         {                                              \
                            (IDXINFOPTR)->isPrimary  = true ;            \
                            (IDXINFOPTR)->isUnique   = true ;            \
                         } \
                         if (key->flags & HA_NOSAME )                    \
                                      (IDXINFOPTR)->isUnique   = true ;



#define   TRY_LOCK_SYSTEM_DB      \
                   do{                                               \
                    DbRetVal lck_rv = systemDatabase_->getDatabaseMutex();\
                      if (BackEnd::OK != lck_rv)                  \
                    {                                            \
                     push_warning(current_session, COLLAPSAR_ERROR::WARN_LEVEL_WARN,\
                             ER_CANT_GET_DB_LOCK, ER(ER_CANT_GET_DB_LOCK)); \
                          return  1;                             \
                    }                                            \
                    } while(0)


DatabaseManagerImpl::~DatabaseManagerImpl()
{
    delete tMgr_;
    delete lMgr_;
}

void DatabaseManagerImpl::createLockManager()
{
    if ( !lMgr_ )
    lMgr_ = new LockManager(systemDatabase_,db_,this);
    return;
}

void DatabaseManagerImpl::createTransactionManager()
{
    if (!tMgr_)
    {
        tMgr_ = new TransactionManager();
        tMgr_->setFirstTrans(systemDatabase_->getSystemDatabaseTrans(0));
    }

    return;
}
void DatabaseManagerImpl::setProcSlot()
{
    if(current_session)
      current_session->setProcSlot(procSlot);
}


DbRetVal DatabaseManagerImpl::openSystemDatabase()
{
    DbRetVal rv = openDatabase(SYSTEMDB);

    if (rv != OK) return rv;
    systemDatabase_ = db_;
    db_ = NULL;
    return OK;
}

DbRetVal DatabaseManagerImpl::closeSystemDatabase()
{
    Database *l_db  = db_;
    //make them to point to system database file descriptor
    //and database pointer
    db_ = systemDatabase_;
    closeDatabase();
    db_ = l_db;
    return OK;
}

/*
 * SYNOPSIS initDataBase()
 *
 * @parameter  void
 * @return 0   Success
 *         1   Error
 */

uint32_t DatabaseManagerImpl::initDataBase() {

  errmsg_printf(ERRMSG_LVL_INFO, _("DatabaseManager starts initDataBase\n"));
  if (g_systemDatabase) return 0 ; /*It is already initialized */
  DbRetVal rv  ;
  DatabaseManagerImpl dbimpl ;

  if (OK != dbimpl.createDatabase(SYSTEMDB, Conf::config.getMaxSysDbSize()))
  {
    errmsg_printf(ERRMSG_LVL_ERROR, _("DatabaseManager failed to create SYSTEMDB database.\n"));
    return (1) ;
  }
  g_systemDatabase  = dbimpl.db();
  rv = g_systemDatabase->getDatabaseMutex();
   if (OK != rv)
  {
       delete g_systemDatabase ; g_systemDatabase=0;
       return 1;
  }
  g_systemDatabase->createAllCatalogTables() ;

  if( g_systemDatabase->initTblNameHash() ||
      g_systemDatabase->initTblFieldHash())
  {
    errmsg_printf(ERRMSG_LVL_ERROR,
         _("System database failed to initialize internal hash table.\n"));
    g_systemDatabase->releaseDatabaseMutex();
    delete g_systemDatabase ;
    g_systemDatabase = NULL;
    return 1 ;
  }

  void *ret = NULL;
   //Allocate space for the lock hash bucket
   ret = g_systemDatabase->allocLockHashBuckets();
   if (NULL == ret)
  {
     g_systemDatabase->releaseDatabaseMutex();
     errmsg_printf(ERRMSG_LVL_ERROR,
                   _("Fatal error.System database fail to allocate hash buckets.\n"));
     delete g_systemDatabase ;
     g_systemDatabase = 0;
     return 1;
  }
  g_systemDatabase->releaseDatabaseMutex();
   //create user database
   dbimpl.setDb(0) ;
   rv = dbimpl.createDatabase(USERDB, Conf::config.getMaxDbSize());
    if (OK != rv)
   { errmsg_printf(ERRMSG_LVL_ERROR,
                  _("Fatal error.Fail to create user database.\n"));
     delete g_systemDatabase ;
     g_systemDatabase = 0;
     return (1) ;
   }
   g_database = dbimpl.db() ;
  /*assign systemdatabase pointer */
   ProcessManager::systemDatabase = g_systemDatabase  ;
   dbimpl.setSysDb(g_systemDatabase);
   g_systemDatabase->initLockHashBucketTrees(dbimpl);
   return (0) ;
}


uint32_t DatabaseManagerImpl::endDataBase(){

  DatabaseManagerImpl dbimpl ;
  dbimpl.setSysDb(g_systemDatabase);
  dbimpl.setDb(g_database);
  /*We ignore all the error*/
  dbimpl.deleteDatabase(SYSTEMDB);
  dbimpl.deleteDatabase(USERDB);
  g_systemDatabase = g_database = 0;

  return (0) ;
}

#define         TWOGMB                    2147483648

struct mem_param{
  size_t  therange;
  void*   address ;
};

static void memset64( void *_dest, uint64_t _value, size_t _size )
{
  size_t i;
  for( i = 0; i < (_size & (~7)); i+=8 )
  {
    memcpy( ((char*)_dest) + i, &_value, 8 );
  }
  for( ; i < _size; i++ )
  {
    ((char*)_dest)[i] = ((char*)&_value)[i&7];
  }
}

static void* memworker(void* inparam){
  struct mem_param * _param = (struct mem_param*)inparam;
  void* _addr   = _param->address;
  size_t _range = _param->therange;
  memset64(_addr,(uint64_t)0,_range);
  pthread_exit(NULL);
}
static void fastZEROdatabase(void* starting_point,size_t _size)
{
  pthread_t tid[2];
  char*  pt[2];
  struct mem_param _param[2] ;

  pthread_attr_t attr ;
  /* Initialize and set thread detached attribute */
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
  size_t range = _size >> 1 ;
  for(uint64_t i=0;i<2; ++i){
    pt[i] =((char*)starting_point+(uint64_t)(i*range));
    _param[i].therange =  range;
    _param[i].address  = (void*)pt[i];
    pthread_create(&tid[i], &attr, memworker, (void *)&_param[i]);
  }
  pthread_attr_destroy(&attr) ;

  for(uint32_t i=0;i<2;++i){
    pthread_join(tid[i],NULL);
  }
}



BackEnd::DbRetVal DatabaseManagerImpl::createDatabase(const char *name,
                                                           size_t size)
{
    if (db_)
    {
      errmsg_printf(ERRMSG_LVL_WARN, _("Database %s is already created.\n"),name);
      return BackEnd::ER_Already;
    }
    caddr_t rtnAddr = (caddr_t) NULL;
    shared_memory_id shm_id = 0;

    char *startaddr = (char*)Conf::config.getMapAddress();
    shared_memory_key key = 0;
    bool b_issysdb   = false ;
    if (!strcmp(name, SYSTEMDB))
    {
      key = Conf::config.getSysDbKey(); b_issysdb = true ;
    }else
    {
      startaddr = startaddr + Conf::config.getMaxSysDbSize();
      key = Conf::config.getUserDbKey();
    }
    errmsg_printf(ERRMSG_LVL_INFO,
                  _("Creating share memory with key [%d].\n"),key);
    shm_id = OS::shm_create(key, size, 0666);
    if (-1 == shm_id)
    {
      if (errno == EEXIST) {
          // Already exists.
          errmsg_printf(ERRMSG_LVL_WARN,
                  _("Warning.The share memory with key [%d] exists...\n"),key);
          shm_id = OS::shm_open(key,size, 0666);
      }
      if( 0> shm_id ) {
        errmsg_printf(ERRMSG_LVL_ERROR,
              _("Fatal error.Fail to open share memory with key [%d] exists.\n"),key);
        return BackEnd::ER_OS;
      }
    }
    void *shm_ptr = OS::shm_attach(shm_id, startaddr, SHM_RND);
    rtnAddr  = (caddr_t) shm_ptr;
    errmsg_printf(ERRMSG_LVL_INFO, _("Share memory address [%p]\n"),rtnAddr);
    if (rtnAddr < 0 || shm_ptr == (char*)0xffffffff || shm_ptr == (char*)HA_POS_ERROR)
    {
      errmsg_printf(ERRMSG_LVL_ERROR,
                  _("Fatal error.The share memory address is invalid!\n"));
      return BackEnd::ER_OS;
    }
    /*Adding very basic check for some system only has very limit default system value*/
    char* endAddr ;
    if (b_issysdb) {
      endAddr = (char*)rtnAddr + Conf::config.getMaxSysDbSize();
    }else{
      endAddr = (char*)rtnAddr + Conf::config.getMaxDbSize();
    }
    errmsg_printf(ERRMSG_LVL_DBUG,
                              _("[%s] [%d] The upper boundary address is %p.\n")
                                ,__FILE__,__LINE__,endAddr);
    if ( endAddr < rtnAddr ){
      /*bizarre*/
      errmsg_printf(ERRMSG_LVL_ERROR,
                   _("Fatal error.The share memory address is invalid!Please refer to user manual for OS kernel parameter tuning\n"));
      return BackEnd::ER_OS;
    }
    errmsg_printf(ERRMSG_LVL_INFO, _("Start to zero database content...\n"));
    if ( size <= TWOGMB )
       memset(shm_ptr, 0, size );
    else
       fastZEROdatabase(shm_ptr,size);

    db_ = new Database();
    errmsg_printf(ERRMSG_LVL_INFO, _("Creating database: [%s]\n"),name);

    db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);
    db_->setDatabaseID(sql_rnd());
    db_->setName(name);
    db_->setMaxSize(size);
    db_->setNoOfChunks((uint32_t)0);
    db_->initAllocDatabaseMutex();
    db_->initDatabaseMutex();
    //or user database do not have transtable and processtable mutex
    if(b_issysdb)
    {
      db_->initTransTableMutex();
      db_->initProcessTableMutex();
    }
    db_->setUniqueChunkID((ulong)((unsigned)time(NULL)<<4)+(ulong)(sql_rnd()&0xffffff));
    //compute the first page after book keeping information
    size_t offset = OS::alignLong(sizeof (DatabaseMetaData));
    //Only for system db chunk array, trans array and proc array will be there
    if (b_issysdb)
    {
      offset += OS::alignLong( MAX_CHUNKS  * sizeof (Chunk));
      offset += OS::alignLong( Conf::config.getMaxProcs()*sizeof(Transaction));
      offset += OS::alignLong( Conf::config.getMaxProcs()*sizeof(ThreadInfo));
    }
    uint64_t multiple = OS::floor(offset / PAGE_SIZE);
    char *curPage = (((char*)rtnAddr) + ((multiple + 1) * PAGE_SIZE));

    db_->setCurrentPage(curPage);
    db_->setFirstPage(curPage);

    return OK;
}

DbRetVal DatabaseManagerImpl::deleteDatabase(const char *name)
{
	errmsg_printf(ERRMSG_LVL_DBUG,
	                      _("[%s]:line [%d] Delete database %s.\n")
	                        ,__FILE__,__LINE__,name);
    shared_memory_id shm_id = 0;
    if (!strcmp(name, SYSTEMDB))
    {
        shm_id = OS::shm_open(Conf::config.getSysDbKey(), 100, 0666);
        OS::shmctl(shm_id, IPC_RMID);
        delete systemDatabase_;
        systemDatabase_ = NULL;
    } else {
        shm_id = OS::shm_open(Conf::config.getUserDbKey(), 100, 0666);
        OS::shmctl(shm_id, IPC_RMID);
        delete db_;
        db_ = NULL;
    }
    return OK;
}

DbRetVal DatabaseManagerImpl::openDatabase(const char *name)
{
    long size = Conf::config.getMaxSysDbSize();
    char *startaddr = (char*)Conf::config.getMapAddress();
    if (0 == strcmp(name , SYSTEMDB))
    {
        if (NULL !=systemDatabase_)
        {
          return BackEnd::ER_Already;
        }
    }
    else
    {
        if (NULL==systemDatabase_)
        {
          return BackEnd::ER_NotOpen;
        }
        size = Conf::config.getMaxDbSize();
        startaddr = startaddr +  Conf::config.getMaxSysDbSize();
    }
    if (NULL != db_)
    {
        return BackEnd::ER_Already;
    }
    //system db should be opened before user database files
    caddr_t rtnAddr = (caddr_t) NULL;

    shared_memory_id shm_id = 0;
    shared_memory_key key = 0;

    if (0 == strcmp(name, SYSTEMDB))
        key = Conf::config.getSysDbKey();
    else
       key = Conf::config.getUserDbKey();


    int ret = ProcessManager::prcMgr_mutex.getLock(-1, false);
    //If you are not getting lock ret !=0, it means somebody else is there.
    //he will close the database.
    if (ret != 0)
    {
      return BackEnd::ER_SysInternal;
    }
    void *shm_ptr = NULL;
    bool firstThread = false;

    if ((ProcessManager::noThreads == 0 && 0 == strcmp(name, SYSTEMDB))
       || (ProcessManager::noThreads == 1 && 0 != strcmp(name, SYSTEMDB)) )
    {
       shm_id = OS::shm_open(key, size, 0666);
       if (shm_id == -1 )
       {
           ProcessManager::prcMgr_mutex.releaseLock(-1, false);
           return BackEnd::ER_OS;
       }
       shm_ptr = OS::shm_attach(shm_id, startaddr, SHM_RND);
       if (0 == strcmp(name, SYSTEMDB))
       {
             firstThread = true;
             ProcessManager::sysAddr = (char*) shm_ptr;
       }
       else
       {
              ProcessManager::usrAddr = (char*) shm_ptr;
       }
    } else {
        if (0 == strcmp(name, SYSTEMDB))
              shm_ptr = ProcessManager::sysAddr;
        else
              shm_ptr = ProcessManager::usrAddr;
    }
    ProcessManager::prcMgr_mutex.releaseLock(-1, false);


    rtnAddr  = (caddr_t) shm_ptr;

    if (rtnAddr < 0 || shm_ptr == (char*)0xffffffff)
    {
      return BackEnd::ER_OS;
    }
    db_ = new Database();
    db_->setMetaDataPtr((DatabaseMetaData*)rtnAddr);

    //if (firstThread) ProcessManager::systemDatabase = db_;

    return OK;
}

BackEnd::DbRetVal DatabaseManagerImpl::closeDatabase()
{

    if (NULL == db_)
    {
        //Database is already closed
        return BackEnd::OK;
    }
    //check if this is the last thread to be deregistered
    int ret = ProcessManager::prcMgr_mutex.getLock(-1, false);
    //If you are not getting lock ret !=0, it means somebody else is there.
    //he will close the database.
    if (ret == 0) {

    if ((ProcessManager::noThreads == 0    &&
        !strcmp((char*)db_->getName(), SYSTEMDB))
        ||(ProcessManager::noThreads == 1 &&
           0 != strcmp((char*)db_->getName(),  SYSTEMDB)) )
      {
        OS::shm_detach((char*)db_->getMetaDataPtr());
      }
    }
    ProcessManager::prcMgr_mutex.releaseLock(-1, false);
    //logFinest(logger, "Closed database");
    delete db_;
    db_ = NULL;
    return BackEnd::OK;
}


//Assumes that system database mutex is taken before calling this.
Chunk* DatabaseManagerImpl::createUserChunk(size_t size)
{
    //Allocate new node in system database to store
    Chunk *chunk = getSystemTableChunk(UserChunkTableId);

    BackEnd::DbRetVal rv = BackEnd::OK;
    void *ptr = chunk->allocate(systemDatabase_, &rv);
    if (!ptr)
    {
      db_error(ER_OUTOFMEMORY, MYF(0), size);
      return NULL;
    }
    Chunk *chunkInfo = (Chunk*)ptr;
    ///chunk init
    chunkInfo->init();

    if (size) chunkInfo->setSize(size);
    if (chunkInfo->allocSize_ > PAGE_SIZE)
        chunkInfo->curPage_ = db_->getFreePage(chunkInfo->allocSize_);
    else
        chunkInfo->curPage_ = db_->getFreePage();

    if (!chunkInfo->curPage_)
    {
        chunkInfo->destroyMutex();
        chunk->free(db_, ptr);
        return NULL;
    }
    PageInfo* firstPageInfo = ((PageInfo*)chunkInfo->curPage_);
    if (chunkInfo->allocSize_ > PAGE_SIZE)
    {
        size_t multiple = OS::floor(chunkInfo->allocSize_ / PAGE_SIZE);
        size_t offset = ((multiple + 1) * PAGE_SIZE);
        firstPageInfo->setPageAsUsed(offset);
    }
    else
    {
        firstPageInfo->setPageAsUsed(chunkInfo->allocSize_);
        char *data = ((char*)firstPageInfo) + sizeof(PageInfo);
        *(int*)data =0;
    }
    if (!size)
    {
        VarSizeInfo *varInfo =
                  (VarSizeInfo*)(((char*)firstPageInfo) + sizeof(PageInfo));
        varInfo->isUsed_ = 0;
        varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
    }
    chunkInfo->firstPage_ = chunkInfo->curPage_;

    if (!size)
        chunkInfo->setAllocType(VariableSizeAllocator);
    else
        chunkInfo->setAllocType(FixedSizeAllocator);

    //TODO::Generate chunkid::use tableid
    chunkInfo->setChunkID(db_->getUniqueIDForChunk());
    db_->incrementChunk();

    return chunkInfo;
}

//Assumes that system database mutex is taken before calling this.
Chunk* DatabaseManagerImpl::createUserIndexChunk(size_t size)
{
    //Allocate new node in system database to store
    Chunk *chunk = getSystemTableChunk(UserIndexChunkId);

    BackEnd::DbRetVal rv = BackEnd::OK;
    void *ptr = chunk->allocate(systemDatabase_, &rv);
    if (!ptr)
    {
      db_error(ER_OUTOFMEMORY, MYF(0), size);
      return NULL;
    }
    Chunk *chunkInfo = (Chunk*)ptr;
    ///chunk init
    chunkInfo->init();

    if (size) chunkInfo->setSize(size);
    if (chunkInfo->allocSize_ > PAGE_SIZE)
        chunkInfo->curPage_ = db_->getFreePage(chunkInfo->allocSize_);
    else
        chunkInfo->curPage_ = db_->getFreePage();

    if (!chunkInfo->curPage_)
    {
        chunkInfo->destroyMutex();
        chunk->free(db_, ptr);
        return NULL;
    }
    PageInfo* firstPageInfo = ((PageInfo*)chunkInfo->curPage_);
    if (chunkInfo->allocSize_ > PAGE_SIZE)
    {
        size_t multiple = OS::floor(chunkInfo->allocSize_ / PAGE_SIZE);
        size_t offset = ((multiple + 1) * PAGE_SIZE);
        firstPageInfo->setPageAsUsed(offset);
    }
    else
    {
        firstPageInfo->setPageAsUsed(chunkInfo->allocSize_);
        char *data = ((char*)firstPageInfo) + sizeof(PageInfo);
        *(int*)data =0;
    }
    if (!size)
    {
        VarSizeInfo *varInfo =
                  (VarSizeInfo*)(((char*)firstPageInfo) + sizeof(PageInfo));
        varInfo->isUsed_ = 0;
        varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);
    }
    chunkInfo->firstPage_ = chunkInfo->curPage_;

    if (!size)
        chunkInfo->setAllocType(VariableSizeAllocator);
    else
        chunkInfo->setAllocType(FixedSizeAllocator);

    //TODO::Generate chunkid::use tableid
    chunkInfo->setChunkID(db_->getUniqueIDForChunk());
    db_->incrementChunk();

    return chunkInfo;
}

//Assumes that system database mutex is get before calling this function.
BackEnd::DbRetVal DatabaseManagerImpl::deleteUserChunk(Chunk *chunk)
{
    if(!chunk) return OK ;// just return .
    //Go to the pages and set them to notUsed
    Page *page = chunk->firstPage_;
    PageInfo* pageInfo = ((PageInfo*)page);
    //Here...sure that at least one page will be there even no tuples
    //are inserted.so not checking if pageInfo == NULL

      while( pageInfo->nextPage_ != NULL)
    {
        PageInfo *prev = pageInfo;
        pageInfo = (PageInfo*)(pageInfo->nextPage_);
        //sets pageInfo->isUsed_ = 0 and pageInfo->hasFreeSpace_ = 0
        //and initializes the page content to zero
        if( !pageInfo->nextPageAfterMerge_ )
          memset(prev, 0, PAGE_SIZE);
        else
        {
          size_t size = (char*) pageInfo->nextPageAfterMerge_ -(char*) pageInfo;
          memset(prev, 0, size);
        }
    }
    //The above loop wont execute for the last page
    //and for the case where table has only one page
    if(!pageInfo->nextPageAfterMerge_)
        memset(pageInfo, 0, PAGE_SIZE);
    else
    {
        size_t size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
        memset(pageInfo, 0, size);
    }
    chunk->reset();

    db_->decrementChunk();

    Chunk *userChunk = getSystemTableChunk(UserChunkTableId);

    userChunk->free(systemDatabase_,chunk);
    errmsg_printf(ERRMSG_LVL_DBUG,
                       _("[%s]: line [%d] Deleting user chunk: [%p].\n"),
                         __FILE__,__LINE__,chunk);
    return BackEnd::OK;
}

//Assumes that system database mutex is get before calling this function.
BackEnd::DbRetVal DatabaseManagerImpl::deleteUserIndexChunk(Chunk *chunk)
{
    if(!chunk) return OK ;// just return .
    //Go to the pages and set them to notUsed
    Page *page = chunk->firstPage_;
    PageInfo* pageInfo = ((PageInfo*)page);
    //Here...sure that at least one page will be there even no tuples
    //are inserted.so not checking if pageInfo == NULL

      while( pageInfo->nextPage_ != NULL)
    {
        PageInfo *prev = pageInfo;
        pageInfo = (PageInfo*)(pageInfo->nextPage_);
        //sets pageInfo->isUsed_ = 0 and pageInfo->hasFreeSpace_ = 0
        //and initializes the page content to zero
        if( !pageInfo->nextPageAfterMerge_ )
          memset(prev, 0, PAGE_SIZE);
        else
        {
          size_t size = (char*) pageInfo->nextPageAfterMerge_ -(char*) pageInfo;
          memset(prev, 0, size);
        }
    }
    //The above loop wont execute for the last page
    //and for the case where table has only one page
    if(!pageInfo->nextPageAfterMerge_)
        memset(pageInfo, 0, PAGE_SIZE);
    else
    {
      size_t size = (char*) pageInfo->nextPageAfterMerge_ - (char*) pageInfo;
      memset(pageInfo, 0, size);
    }
    chunk->reset();

    db_->decrementChunk();

    Chunk *userIdxChunk = getSystemTableChunk(UserIndexChunkId);
    userIdxChunk->free(systemDatabase_,chunk);
    errmsg_printf(ERRMSG_LVL_DBUG,
                       _("[%s]: line [%d] Deleting user chunk: [%p].\n"),
                         __FILE__,__LINE__,chunk);
    return BackEnd::OK;
}
/*
   SYNOPSIS:
   createTable
   @param   tblname,     The table name
   @param   Table&       The reference to the table
   @param   create_info  pointer to the HA_CREATE_INFO
   @param   internal_table

    @return ER_OUTOFMEMORY -> Unable to create chunk. No memory
    @return -2 -> Unable to update the catalog tables
    @return -31 ->  ER_TBL_CREATION .
*/
BackEnd::DbRetVal DatabaseManagerImpl::createTable(const char *tblname,
                                                   Table& tbl,
                                                   HA_CREATE_INFO * create_info,
                                                   bool internal_table )
{
    BackEnd::DbRetVal rv = OK;
    Session* _session =  current_session ;
    uint32_t fldCount = tbl.sizeFields() ;
    if(!fldCount)
    {
      db_message(ER_TABLE_MUST_HAVE_COLUMNS, ER(ER_TABLE_MUST_HAVE_COLUMNS),
                   MYF(0));
      return (DbRetVal)HA_ERR_TABLE_MUST_HAVE_COLUMNS;
    }
    //If total field count is less than 32, then 1 integer is used to store all
    //null information,
    //if it is more then 1 char is used to store null information
    //of each field
    //This is to done to reduce cpu cycles for small tables
    size_t addSize = 0;
    if (fldCount < 31) addSize = 4;
      else addSize = OS::align(fldCount);

    size_t sizeofTuple = OS::align(tbl.s->stored_rec_length+ addSize);

    rv = systemDatabase_->getDatabaseMutex();
    if ( BackEnd::OK != rv )
    {
      push_warning(_session, COLLAPSAR_ERROR::WARN_LEVEL_WARN,
                   ER_CANT_GET_DB_LOCK, ER(ER_CANT_GET_DB_LOCK));
      return rv;
    }

    void *tptr  = NULL;
    void *chunk = NULL;

    //check whether table already exists
    CatalogTableTABLE cTable(systemDatabase_);
    rv=checkTableExists(tblname,false);
    if ( rv !=  ER_NotExists) return rv ;

    //create a chunk to store the tuples
    Chunk *ptr = createUserChunk(sizeofTuple);
    if (!ptr)
    {
        systemDatabase_->releaseDatabaseMutex();
        db_error(ER_OUTOFMEMORY, MYF(0), sizeofTuple);
        return (BackEnd::DbRetVal)HA_ERR_OUT_OF_MEM;
    }

    ptr->setChunkName(tblname);
    //add row to TABLE
    int64_t tblID = 0;// unused.

    enum TABLETYPE tbl_type_;

    internal_table ?  tbl_type_ = INTERNAL_TYPE : tbl_type_ = NORMAL_TYPE ;

    // Call   CatalogTableTABLE insert
    rv = cTable.insert(tblname,
                         tblID,
                         sizeofTuple,
                         fldCount,
                         ptr,
                         tptr,
                         (int32_t)strlen(tblname),
                         create_info,
                         tbl_type_);
    if (BackEnd::OK != rv)
    {
        deleteUserChunk(ptr);
        systemDatabase_->releaseDatabaseMutex();
        return (BackEnd::DbRetVal)HA_ERR_SYS_INTERNAL;
    }
    /**
     * Within the block of quick_openTable
     * systemDatabase_->releaseDatabaseMutex()
     * will be called.
     */
    if (quick_openTable(tblname,&tbl,chunk,tptr,(char**)NULL,true) )
    {
        /* on Error*/
        ///delete the CTABLE
        rv = systemDatabase_->getDatabaseMutex();
        if (BackEnd::OK != rv ) {
          push_warning(_session, COLLAPSAR_ERROR::WARN_LEVEL_WARN,
                        ER_CANT_GET_DB_LOCK, ER(ER_CANT_GET_DB_LOCK));
          return rv;
        }
        deleteUserChunk(ptr)             ;
        void *cptr, *ttptr;//Dummy as remove below needs both these OUT params
        cTable.remove(tblname, cptr, ttptr)    ;
        systemDatabase_->releaseDatabaseMutex();
        return BackEnd::ER_TBL_CREATION  ;
    }
    return BackEnd::OK;
}

BackEnd::DbRetVal DatabaseManagerImpl::dropTable(const char *tblname)
{
    void *chunk = NULL;
    void *tptr  = NULL;
    Session* _session = current_session ;
    BackEnd::DbRetVal rv = systemDatabase_->getDatabaseMutex();

      if (BackEnd::OK != rv)
    {
      push_warning(_session, COLLAPSAR_ERROR::WARN_LEVEL_WARN,
                   ER_CANT_GET_DB_LOCK, ER(ER_CANT_GET_DB_LOCK));
      return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;
    }
    //remove the entry in TABLE
    CatalogTableTABLE cTable(systemDatabase_);

    rv = cTable.getChunkAndTblPtr(tblname, chunk, tptr);
    if( OK != rv){
        if (BackEnd::ER_NotFound == rv)
      {
        systemDatabase_->releaseDatabaseMutex();
        return (DbRetVal)HA_ERR_NO_SUCH_TABLE;
      }else
      {
        systemDatabase_->releaseDatabaseMutex();
        return (DbRetVal)HA_ERR_INTERNAL_ERROR;
      }
    }
    /// Exclusively lock the chunk.
    rv = lMgr_->getExclusiveLock(chunk, NULL);

      if(rv !=BackEnd::OK)
    {
      systemDatabase_->releaseDatabaseMutex();
      push_warning_printf(_session, COLLAPSAR_ERROR::WARN_LEVEL_WARN,
         ER_CANT_GET_EXCLUSIVE_LOCK, ER(ER_CANT_GET_EXCLUSIVE_LOCK),tblname);
      return rv;
    }

    CatalogTableINDEX cIndex(systemDatabase_) ;
    uint32_t noIndexes = cIndex.getNumIndexes(tptr);

    for (uint32_t i =1 ; i<= noIndexes; i++) {
        char *idxName = cIndex.getIndexName(tptr, i);
        dropIndexInt(idxName, tptr, false);
    }
    ((Chunk*)chunk)->print();
    rv = deleteUserChunk((Chunk*)chunk);
    //remove the entries in the FIELD table
    /*
    CatalogTableFIELD cField(systemDatabase_);

    rv = cField.remove(tptr);
    if (BackEnd::OK!= rv) {
       // remove CatalogTableFIELD entry failed
        lMgr_->releaseLock(chunk);
        systemDatabase_->releaseDatabaseMutex();
        return (DbRetVal)HA_ERR_INTERNAL_ERROR;
    }
    errmsg_printf(ERRMSG_LVL_DBUG, _("[%s] : line [%d] Deleted CFIELD.Table name: [%s] \n"),
                      __FILE__,__LINE__,tblname);
    */
    /*Remove CTABLE*/
    rv = cTable.remove(tblname, chunk, tptr);
      if (rv !=BackEnd::OK)
    {
      lMgr_->releaseLock(chunk);
      systemDatabase_->releaseDatabaseMutex();
      return BackEnd::ER_SysInternal;
    }
    errmsg_printf(ERRMSG_LVL_DBUG, _("[%s] : line [%d] Deleted CTABLE [%s] \n"),
                  __FILE__,__LINE__,tblname);


    Chunk *chunkNode=systemDatabase_->getSystemDatabaseChunk(UserChunkTableId);
    chunkNode->free(systemDatabase_, (Chunk *) chunk);

    /// Release the exclusive Lock
    rv = lMgr_->releaseLock(chunk);
    systemDatabase_->releaseDatabaseMutex();
    return BackEnd::OK;
}

/**
 * openTable
 * @parameter    name const char*
 * @parameter    tbl  Table*
 * @parameter    b_creatIdx  bool  the default value is true,
 *               creating the index.
 * @return onError    return none zero value
 *         success    return 0
 */
int DatabaseManagerImpl::openTable (const char *name,
                                    Table* tbl,
                                    bool b_creatIdx )
{

#define TRY_TO_CREATE_INX(HASHIDXINFO)                                  \
              do{                              \
                  BackEnd::DbRetVal _ret =    createIndex(              \
                                              _idxname.c_str(),         \
                                              *tbl,                     \
                                               (HASHIDXINFO),           \
                                               false ) ;                \
                   if (BackEnd::OK != _ret  )                           \
                 {                                                      \
                     if ( _ret !=(DbRetVal) HA_ERR_FOUND_DUPP_KEY )     \
                        systemDatabase_->releaseDatabaseMutex();        \
                      return 1 ;                                        \
                 }                                                     \
              } while (0)

    if (!tbl )  return  1 ;
        DbRetVal rv = systemDatabase_->getDatabaseMutex();
    if (OK != rv) {
      push_warning(current_session, COLLAPSAR_ERROR::WARN_LEVEL_WARN,
                   ER_CANT_GET_DB_LOCK, ER(ER_CANT_GET_DB_LOCK));
      return  1;
    }
    //to store the chunk pointer of table
    void *chunk = NULL;
    //to store the tuple pointer of the table
    void *tptr  =NULL;


    CatalogTableTABLE cTable(systemDatabase_);

    rv = cTable.getChunkAndTblPtr(name, chunk, tptr);
    if ( OK != rv)
    {
      errmsg_printf(ERRMSG_LVL_DBUG, _("[%s] : line [%d] open table [%s],but table does not exist. \n"),
                    __FILE__,__LINE__,name);
      systemDatabase_->releaseDatabaseMutex();
      return  HA_ERR_NO_SUCH_TABLE;
    }
    tbl->setDB(db_);
    tbl->setSystemDB(systemDatabase_);
    tbl->setLockManager(lMgr_);
    tbl->setTrans(ProcessManager::getThreadTransAddr(procSlot));

    CatalogTableINDEX cIndex(systemDatabase_) ;
    CTABLE *tTuple = (CTABLE*)tptr;
    /// set the table chunk pointer.
    tbl->setChunkPtr(tTuple->chunkPtr_) ;
    tbl->setCTABLEPtr(tTuple);
    /// numIndexes_
    int _keynum = tbl->numIndexes_  =  tbl->s->keys    ;
    tTuple->keys   = _keynum            ;
    tTuple->currently_disabled_keys = 0 ;

     int primary_key_no = -1 ;

     primary_key_no= (tbl->s->primary_key != MAX_KEY ?
                     (int) tbl->s->primary_key : -1);

      for (int i =0 ; i < _keynum   ; i++ )
    {
        //Todo: We use the Hash Index first.
        HashIndexInfo *hIdxInfo = new HashIndexInfo();
        /* fill in the tbl name */
        strcpy(hIdxInfo->tableName,name );
        KEY*  key;
        uint  n_fields   = 0 ;
        KEY_PART_INFO* key_part;
        // Key pointer
        key = tbl->key_info + i;
        n_fields = key->key_parts;
        hIdxInfo->flag= (key->flags & (HA_NOSAME | HA_NULL_ARE_EQUAL));
        tbl->idxInfoList.append(hIdxInfo);
        std::string _idxname;
        // We have more than one field.
        for (uint j = 0; j < n_fields; j++) {

              key_part = key->key_part + j      ;
              Field *field= key_part->field     ;

              _idxname= std::string(name)+std::string("##idx##")+std::string(key->name) ;
              if( j>0){
                char buffer [33];
                int10_to_str((int32_t)j,buffer,10);
               _idxname += std::string("-@@")+std::string(buffer) ;
              }
              HashIndexInfo*   hashidxinfo = hIdxInfo ;
              if (0 == j) {
                // it is the 1st one, then we use the existing hIdxInfo
                FILL_IN_INDX_INFO(hashidxinfo)    ;
                if ( true == b_creatIdx && tbl->mb_HashOnEachField)
                {
                   TRY_TO_CREATE_INX(hashidxinfo)    ;

                   TRY_LOCK_SYSTEM_DB   ;
                }//
               }else{
                // j >0
                HashIndexInfo *__hIdxInfo = new HashIndexInfo();
                /// The parent pointer.
                __hIdxInfo->parent          =   hIdxInfo       ;
                __hIdxInfo->indexPtr        =   0              ;
                __hIdxInfo->flag = __hIdxInfo->parent->flag    ;
                FILL_IN_INDX_INFO(__hIdxInfo)    ;
                /// The sub leaf pointer.
                hIdxInfo->sublist.append(__hIdxInfo)  ;
                //__hIdxInfo->noOfBuckets   =
                //CatalogTableINDEX::getNoOfBuckets(table->indexPtr_[i]);
                if ( b_creatIdx && tbl->mb_HashOnEachField) {
                   TRY_TO_CREATE_INX(__hIdxInfo);
                   TRY_LOCK_SYSTEM_DB ;
                 }
             }// else
         } // for (int j = 0; j < n_fields; j++)
         if ( b_creatIdx && !tbl->mb_HashOnEachField)
        {
          TRY_TO_CREATE_INX(hIdxInfo);
          TRY_LOCK_SYSTEM_DB   ;
        }//
         if( !b_creatIdx)
        {
          strcpy(hIdxInfo->idxName,_idxname.c_str() );
        }
        hIdxInfo->indexPtr =(unsigned char*) cIndex.getIndexPtrByName(tTuple,_idxname.c_str());
    }  //for (int i =0 ; i < tbl.s->keys   ; i++ )

    tbl->increaseBEOpenCount () ;
    systemDatabase_->releaseDatabaseMutex();
    return (0) ;
}

/**
 * SYNOPSIS checkTableExists
 * @param name     char*  the table name
 * @param b_getlock  whether to get database mutex.
 * @return  on System Error             BackEnd::ER_SysInternal
 *          the table exists            BackEnd::ER_Already
 *          the table does not exist    BackEnd::ER_NotExists
 *
 */
BackEnd::DbRetVal DatabaseManagerImpl::checkTableExists(const char *tblname,bool b_getlock)
{
    /*try to get the lock*/
    Session* _session =  current_session ;
    if (b_getlock){
      DbRetVal rv = systemDatabase_->getDatabaseMutex();
      if (OK != rv) {
       push_warning(_session, COLLAPSAR_ERROR::WARN_LEVEL_WARN,
                    ER_CANT_GET_DB_LOCK, ER(ER_CANT_GET_DB_LOCK));
        return  BackEnd::ER_SysInternal;
      }
    }
    void *tptr  =NULL;
    void *chunk = NULL;

     //check whether table already exists
    CatalogTableTABLE cTable(systemDatabase_);
    cTable.getChunkAndTblPtr(tblname, chunk, tptr);
     if (tptr)
    {
       if (b_getlock)systemDatabase_->releaseDatabaseMutex();
       return BackEnd::ER_Already;
    }
     if (b_getlock) systemDatabase_->releaseDatabaseMutex();
    return BackEnd::ER_NotExists ;
}

#define QUCIK_CREATE_IDEX(IDXNAME , HASHINFOPTR)   do   {     \
            HashIndexInfo * __hashidxinfo =HASHINFOPTR ;    \
                   /* fill in the tbl name */               \
            strcpy(__hashidxinfo->tableName,name );          \
            __hashidxinfo->flag =(key->flags & (HA_NOSAME | HA_NULL_ARE_EQUAL));\
            if ( BackEnd::OK != quick_createHashIndex_wrapper(           \
                                      IDXNAME,  /*index name*/      \
                                      /*field name*/                         \
                                      key_part->field->field_name ,          \
                                      *tbl,                                  \
                                      __hashidxinfo,                          \
                                      chunk,                                 \
                                      tptr ,                                 \
                                      fptr                                   \
                                      ) )                                    \
             {                                                       \
              errkey =i;                                             \
              errmsg_printf(ERRMSG_LVL_DBUG, _("[%s] : line [%d] Failed to create index [%s]\n"),\
                                                     __FILE__,__LINE__,(IDXNAME));    \
                       return 1 ;       \
             }                                                       \
           }while(0)


/**
 * SYNOPSIS: quick_openTable
 *
 *
 * @note:  before calling quick_openTable,
 *         systemDatabase_'s DatabaseMutex must be acquired.
 *         on any error, releaseDatabaseMutex() will be called.
 */
int DatabaseManagerImpl::quick_openTable(const char *name,
                                         Table* tbl,
                                         void*  chunk,
                                         void*  tptr ,
                                         char** fptr,
                                         bool b_creatIdx)
{
    if ( !tbl )  return (1) ;

    tbl->setDB(db_);
    tbl->setSystemDB(systemDatabase_);
    tbl->setLockManager(lMgr_);
    tbl->setTrans(ProcessManager::getThreadTransAddr(procSlot));

    CTABLE *tTuple = (CTABLE*)tptr;
    //   table's chunk pointer
    tbl->setChunkPtr(tTuple->chunkPtr_) ;
    tbl->setCTABLEPtr(tTuple);

    if(tbl->getOwner())
        strcpy(tTuple->tblowner,tbl->getOwner());
     /// numIndexes_
    uint32_t num_idx = tbl->numIndexes_  =  tbl->s->keys    ;
    tTuple->keys                    = num_idx ;
    tTuple->currently_disabled_keys = 0 ;
    // todo: variable_size
    tTuple->fixed_data_length  = tTuple->reclength  = tbl->s->reclength;
    tTuple->fixed_column_count = tbl->sizeFields();
    uint32_t auto_key= 0, auto_key_type= 0;

     int primary_key_no = -1 ;     errkey =-1 ;

     primary_key_no= (tbl->s->primary_key != MAX_KEY ?
                                          (int) tbl->s->primary_key : -1);

      for (uint32_t i =0 ; i < num_idx   ; i++ )
    {
        // We use the Hash Index first.
        HashIndexInfo *hIdxInfo = new HashIndexInfo();
        KEY*    key             ;
        uint    n_fields = 0    ;
        KEY_PART_INFO*  key_part =0;
        // Key pointer
        key = tbl->key_info + i;
        hIdxInfo->flag= (key->flags & (HA_NOSAME | HA_NULL_ARE_EQUAL));
        n_fields = key->key_parts;
        tbl->idxInfoList.append(hIdxInfo);
        std::string _idxname;
          // We have more than one field.
          for (uint j = 0; j < n_fields; j++)
        {
            key_part = key->key_part + j   ;
            Field *field= key_part->field  ;

            if (field->flags & AUTO_INCREMENT_FLAG &&
                    tbl->found_next_number_field &&
                    i == tbl->s->next_number_index)
           {
              /*
               Store key number and type for found auto_increment key
               We have to store type as seg->type can differ from it
              */
               auto_key= i+ 1;
               auto_key_type= field->key_type();
               tTuple->auto_key = auto_key ;
               tTuple->auto_key_type=auto_key_type;
               tTuple->auto_key_offset = key_part->offset ;
           }

            _idxname=std::string(name)+std::string("##idx##")+std::string(key->name) ;
            if( j>0 ) {
              char buffer [33];
              int10_to_str((int32_t)j,buffer,10);
              _idxname += std::string("-@@")+std::string(buffer) ;
            }
            HashIndexInfo*   hashidxinfo = hIdxInfo ;
            if (0 == j) {
               // it is the 1st one, then we use the existing hIdxInfo
               FILL_IN_INDX_INFO(hashidxinfo)    ;

               if ( true == b_creatIdx && tbl->mb_HashOnEachField ) {
                     QUCIK_CREATE_IDEX(_idxname.c_str(),hashidxinfo)   ;
                     TRY_LOCK_SYSTEM_DB    ;
                }
            }else{
             // j >0
                HashIndexInfo * _hIdxInfo_ = new HashIndexInfo();
                /// The parent pointer.
                _hIdxInfo_->parent        =   hIdxInfo         ;
                _hIdxInfo_->indexPtr      =   0                ;
                _hIdxInfo_->flag = _hIdxInfo_->parent->flag    ;
                FILL_IN_INDX_INFO(_hIdxInfo_)                  ;
                /// The sub leaf pointer.

                _hIdxInfo_->parent->sublist.append(_hIdxInfo_)  ;

                  if ( true == b_creatIdx && tbl->mb_HashOnEachField) {
                      QUCIK_CREATE_IDEX(_idxname.c_str(),_hIdxInfo_) ;
                      TRY_LOCK_SYSTEM_DB    ;
                  }
             }  // else j > 0 ;

        } // for (int j = 0; j < n_fields; j++)
          if ( true == b_creatIdx && !tbl->mb_HashOnEachField)
        {     /* We put the whole field*/
               QUCIK_CREATE_IDEX(_idxname.c_str(),hIdxInfo);
               TRY_LOCK_SYSTEM_DB   ;
        }//
    }  //for (int i =0 ; i < tbl.s->keys   ; i++ )
    tbl->increaseBEOpenCount () ;
    systemDatabase_->releaseDatabaseMutex();
    return 0 ;
}
/*

//Return values: -1 for table not found
*/
void DatabaseManagerImpl::closeTable(Table *table)
{
    if (!table) return;
    table->decreaseBEOpenCount() ;

    table->setDB(0);
    table->setSystemDB(0);
    table->setLockManager(0);
    table->setTrans(0);
    table->setChunkPtr(0) ;
    table->setCTABLEPtr(0);

    table->TrashIdxInfoList()  ;
}
/**
 *@parameter indName:  the indexname
 *@parameter tbl       Table&
 *
 *
 */
BackEnd::DbRetVal DatabaseManagerImpl::createIndex(const char *indName,
                                                   Table& tbl ,
                                                   IndexInfo *info,
                                                   bool b_wlock)
{
    BackEnd::DbRetVal rv = BackEnd::OK;
    if (!info->isUnique && info->isPrimary)
                           return BackEnd::ER_BadCall;

    if (info->indType == hashIndex)
    {
        //Assumes info is of type HashIndexInitInfo
        HashIndexInfo *hInfo = (HashIndexInfo*) info;
        rv = createHashIndex(indName,
                             tbl, /* table */
                             info,
                             hInfo->tableName,  /* table name */
                             tbl.sizeFields(),
                             hInfo->noOfBuckets,
                             hInfo->isUnique,
                             hInfo->isPrimary,
                             false,
                             b_wlock);
    }/*
    else if (info->indType == treeIndex)
    {
        HashIndexInitInfo *hInfo = (HashIndexInitInfo*) info;
        rv = createTreeIndex(indName,
                             tbl, // table
                             info->tableName,
                             info->list,
                             hInfo->bucketSize,
                             info->isUnique,
                             info->isPrimary);

    }*/
    else {
        return BackEnd::ER_BadCall;
    }
    return rv;
}

/**
 * SYNOPSIS disableIndex
 *
 * @param tbl  the reference to Table
 * @param tptr CTABLE pointer
 * @return: BackEnd::OK    Success.
 *          BackEnd::ER_LockTimeOut  Fail to get the lock.
 *
 */
BackEnd::DbRetVal DatabaseManagerImpl::disableIndex(Table& , void* tptr)
{
    BackEnd::DbRetVal rv;
    /* get the mutex.*/
    rv = systemDatabase_->getDatabaseMutex();
    if (BackEnd::OK != rv ) {
        return rv;
    }
   /*TODO: Check if the tTuple is still valid*/
    CTABLE *tTuple = (CTABLE*)tptr;
    tTuple->keys   = tTuple->currently_disabled_keys;
    tTuple->keys   = 0 ;
    systemDatabase_->releaseDatabaseMutex();
    return OK ;
}

/**
 * SYNOPSIS  enableIndex
 *
 * @param tbl
 * @param tptr
 * @return
 */
BackEnd::DbRetVal  DatabaseManagerImpl::enableIndex(Table&, void* tptr)
{
    BackEnd::DbRetVal rv;
    /* get the mutex.*/
    rv = systemDatabase_->getDatabaseMutex();
    if (BackEnd::OK != rv ) {
        return rv;
    }
   /*TODO: Check if the tTuple is still valid*/
    CTABLE *tTuple = (CTABLE*)tptr;
    tTuple->currently_disabled_keys = tTuple->keys ;
    tTuple->keys   = 0;

    systemDatabase_->releaseDatabaseMutex();
    return OK ;
}
/**
 * SYNOPSIS
 *     is_indexes_are_disabled
 * @param tbl
 * @param tptr
 * @return  BackEnd::OK success
 */
BackEnd::DbRetVal DatabaseManagerImpl::is_indexes_are_disabled(Table&, void* tptr)
{
	BackEnd::DbRetVal rv;
    /* get the mutex.*/
    rv = systemDatabase_->getDatabaseMutex();
    if (BackEnd::OK != rv ) {
      return rv;
    }
   /*TODO: Check if the tTuple is still valid*/
    CTABLE *tTuple = (CTABLE*)tptr;

    if(! tTuple->keys && tTuple->currently_disabled_keys)
    {
      rv = OK ;
    }else  rv = ER_IDX_ENABLED;
    systemDatabase_->releaseDatabaseMutex();
    return rv ;
}
/*
 * SYNOPSIS  createHashIndex
 *
 *
 *
 *@return  OK success
*/
BackEnd::DbRetVal DatabaseManagerImpl::createHashIndex
    (const char *indName,  /*  the idx name */
     Table& tbl         ,  /*  the table    */
     IndexInfo *  indxptr ,
     const char * tblName, /*  the table name */
     uint32_t  fldsize,
     int bucketSize,       /*  the bucketsize */
     bool isUnique,
     bool ,
     bool  ,
     bool b_wlock  /* if we lock the database mutex*/)
{
    //validate the bucket size
    if (bucketSize < 100)
    {
        return BackEnd::ER_BadRange;
    }
    if (!fldsize)
      return BackEnd::ER_BadCall;


    void *tptr  = NULL;
    void *chunk = NULL;
    DbRetVal rv     ;
    if (b_wlock ) {
       rv = systemDatabase_->getDatabaseMutex();
      if (BackEnd::OK != rv)
      {
        Session* _session = current_session ;
        push_warning(_session, COLLAPSAR_ERROR::WARN_LEVEL_WARN,
                     ER_CANT_GET_DB_LOCK, ER(ER_CANT_GET_DB_LOCK));
        return BackEnd::ER_SysInternal;
      }
    }
     ///check whether table exists
    CatalogTableTABLE cTable(systemDatabase_);
    cTable.getChunkAndTblPtr(tblName, chunk, tptr);
      if (! tptr)
    {
        if (b_wlock )systemDatabase_->releaseDatabaseMutex();
          return BackEnd::ER_NotExists;
    }
    //create chunk to store the meta data of the index created
    //for latches and bucket pointers


    Chunk* chunkInfo = createUserChunk(bucketSize * sizeof(Bucket));
    if (!chunkInfo)
    {
      if (b_wlock )systemDatabase_->releaseDatabaseMutex();
        return BackEnd::ER_SysInternal;
    }
    chunkInfo->setChunkName(indName);
    //create memory for holding the bucket pointers
    void *buckets = chunkInfo->allocate(db_, &rv);
    if (!buckets)
    {
        deleteUserChunk(chunkInfo);
        if (b_wlock )systemDatabase_->releaseDatabaseMutex();
        return rv;
    }
    Bucket *buck = (Bucket*) buckets;

    initHashBuckets(buck, bucketSize);

    //create chunk to store the hash index nodes
    Chunk* hChunk = createUserChunk(sizeof(HashIndexNode));
    if (!hChunk)
    {
        destroyHashBuckets(buck, bucketSize);
        deleteUserChunk(chunkInfo);
        if (b_wlock )systemDatabase_->releaseDatabaseMutex();
        return BackEnd::ER_SysInternal;
    }
     hChunk->setChunkName(indName);
    //add row to CINDEX
    void *tupleptr = NULL;
    CatalogTableINDEX cIndex(systemDatabase_);
    rv = cIndex.insert(indName,
                       tptr,
                       fldsize,
                       isUnique,
                       chunkInfo,
                       bucketSize,
                       hChunk,
                       tupleptr);
    if (BackEnd::OK != rv)
    {
        //delete[] fptr;
        destroyHashBuckets(buck, bucketSize);
        deleteUserChunk(hChunk);
        deleteUserChunk(chunkInfo);
        if (b_wlock )systemDatabase_->releaseDatabaseMutex();
        return BackEnd::ER_SysInternal;
    }
    /*!!!  We need to release it anyway.  !!!*/
    systemDatabase_->releaseDatabaseMutex();

    //TODO:: Take table lock
    if (! tbl.numTuples()) {
        return BackEnd::OK;
    }

    /*!!!! We assume that only one fieldname when we are using
           hash per field
    */
           /// Find the right key name;
           /// found it.
           HashIndexInfo* hashindxptr = (HashIndexInfo*)indxptr;
           hashindxptr->indexPtr =(unsigned char*) tupleptr        ;
           hashindxptr->buckets  = buck            ;
           if ( !tbl.mb_HashOnEachField)
           {
             /* if mb_HashOnEachField == false,then Hash on compound Field.
              * we just call insertIndexNodeToTable
              */
             rv = insertIndexNodeToTable(chunk,tbl,tupleptr,hashindxptr);
             if(OK != rv )
             {
                 dropIndex(indName,tptr);
                 return rv ;
             }
           }else {
              if ( indxptr->sublist.size() == 0 )
             {
                 rv =insertIndexNodeToTable(chunk,tbl,tupleptr,hashindxptr);
                 if(OK != rv )
                 {
                   // DROP index ;
                     dropIndex(indName,tptr);
                     return rv ;
                 }
              //}
             }else {
                /* We have sub list; Hash on each field */
                BackEnd::ListIterator iter = indxptr->sublist.getIterator() ;

                while(iter.hasElement()) {

                  HashIndexInfo *hidxinfo =(HashIndexInfo*)(iter.nextElement());
                  hidxinfo->indexPtr    = (unsigned char*)tupleptr        ;
                  rv = insertIndexNodeToTable(chunk,tbl,tupleptr,hidxinfo);
                  if(OK != rv)
                  {
                    dropIndex(indName,tptr);
                    return rv;
                  }
                }//while (iter.hasElement())
             }// if ( tbl.idxInfo[i].sublist.size() == 0 )
         }// else
    return BackEnd::OK;
}




BackEnd::DbRetVal DatabaseManagerImpl::quick_createHashIndex_wrapper
                                  (const char *indName,
                                   const char *fldName,
                                   Table& tbl,
                                   HashIndexInfo *hashidxinfo,
                                   void*  chunk,
                                   void*  tptr ,
                                   char** fptr
                                  )
{

   void *tupleptr = NULL;
    /// 1st, We call quick_createHashIndex_part1
    BackEnd::DbRetVal idxrev = quick_createHashIndex_part1
                                    (indName   ,  /*index name*/
                                     fldName,     /*field name*/
                                     tbl,
                                     hashidxinfo,
                                     chunk      ,
                                     tptr       ,
                                     fptr       ,
                                     tupleptr     /* the out parameter*/);

    systemDatabase_->releaseDatabaseMutex();

    if (BackEnd::OK != idxrev )
    {
       return idxrev ;
    }
    /// Check tupleptr again. should not happen???
    if ( !tupleptr)
    {
      return (DbRetVal)HA_ERR_INTERNAL_ERROR;
    }

     /// There will no database lock acquired before calling
     /// quick_createHashIndex_part2
     /// The go to the step part2:
    idxrev = quick_createHashIndex_part2
                                      (indName    ,
                                       fldName    ,
                                       tbl        ,
                                       hashidxinfo,
                                       chunk      ,
                                       tupleptr) ;

    if( idxrev != BackEnd::OK )
    {
         hashidxinfo->buckets = 0;
         return idxrev;
    }

    return BackEnd::OK ;

}

BackEnd::DbRetVal DatabaseManagerImpl::quick_createHashIndex_part1
(
  const char *indName,  /*  the idx name */
  const char *,         /*  the fldName  */
  Table&          ,     /*  the table    */
  HashIndexInfo *info,  /*  Hash index Info */
  void*  ,
  void*  tptr ,
  char** , /* if we lock the database mutex*/
  void*& tupleptr
)
{

#define   DEL_CHUNCK_WITH_ERROR(CHUNK,SIZE)   do  { \
                                   deleteUserChunk(CHUNK);                       \
                                   db_error(ER_OUTOFMEMORY, MYF(0), sizeof(SIZE)); \
                                   return (BackEnd::DbRetVal)ER_OUTOFMEMORY;         \
                                  }while(0)



    size_t bucketSize = info->noOfBuckets      ;

    //validate the bucket size
    if (bucketSize < 100)
    {
      return BackEnd::ER_BadRange;
    }

    //create chunk to store the meta data of the index created
    //for latches and bucket pointers
    Chunk* chunkInfo = createUserChunk(bucketSize * sizeof(Bucket));
    if (!chunkInfo)
    {
       db_error(ER_OUTOFMEMORY, MYF(0), bucketSize * sizeof(Bucket));
       return (BackEnd::DbRetVal)ER_OUTOFMEMORY;
    }
    chunkInfo->setChunkName(indName);
    //create memory for holding the bucket pointers
    BackEnd::DbRetVal rv = BackEnd::ER_NoMemory;;
    void *buckets = chunkInfo->allocate(db_, &rv);
    if (!buckets)
    {
      DEL_CHUNCK_WITH_ERROR(chunkInfo,Bucket);
    }
    Bucket *buck = (Bucket*) buckets;

    initHashBuckets(buck, bucketSize);

    //create chunk to store the hash index nodes
    Chunk* hChunk = createUserChunk(sizeof(HashIndexNode));
    if (!hChunk)
    {
      DEL_CHUNCK_WITH_ERROR(chunkInfo,HashIndexNode);
    }
     hChunk->setChunkName(indName);
    //add row to CINDEX

    CatalogTableINDEX cIndex(systemDatabase_);
    rv = cIndex.insert(indName,
                       tptr   ,
                       (IndexInfo*)info,
                       chunkInfo,
                       hChunk,
                       tupleptr);
    if (BackEnd::OK != rv)
    {
        destroyHashBuckets(buck, bucketSize);
        deleteUserChunk(hChunk);
        deleteUserChunk(chunkInfo);
        return BackEnd::ER_SysInternal;
    }
    /*assign buckets here*/
    info->buckets = buck ;
    errmsg_printf(ERRMSG_LVL_DBUG, _("[%s] : line [%d] Creating HashIndex [%s] "
                                     "on %s with bucket size %d\n"),
                  __FILE__,__LINE__,indName, info->tableName, bucketSize);
    return BackEnd::OK;
}

BackEnd::DbRetVal DatabaseManagerImpl::quick_createHashIndex_part2
                                     (const char *indName,
                                      const char *,
                                      Table& tbl         ,
                                      HashIndexInfo *info,
                                      void* chunk        ,
                                      void* tupleptr
                                      )
{
    if (! tbl.numTuples()) {
        return BackEnd::OK;
    }
    info->indexPtr = (unsigned char*)tupleptr        ;
    BackEnd::DbRetVal ret  = BackEnd::OK;
     if ( !tbl.mb_HashOnEachField)
    {
             if((ret= insertIndexNodeToTable(chunk,tbl,tupleptr,info)))
            {
                   // DROP index ;
                   // Todo:...
               dropIndex(indName,tbl.getChunkPtr());
               return ret ;
            }
     }else {
             if ( info->sublist.size() == 0 )
            {
                   /// found it.
                 if((ret= insertIndexNodeToTable(chunk,tbl,tupleptr,info)))
               {
                     // DROP index ;
                   dropIndex(indName,tbl.getChunkPtr());
                   return ret ;
               }
            }else {
                /* We have sub list */
                  BackEnd::ListIterator iter = info->sublist.getIterator() ;
                  while(iter.hasElement()) {
                  HashIndexInfo * hidxinfo=(HashIndexInfo*)(iter.nextElement());
                  hidxinfo->indexPtr    = (unsigned char*)tupleptr        ;
                    if((ret=insertIndexNodeToTable(chunk,tbl,tupleptr,hidxinfo)))
                  {
                      //Drop Index.
                      dropIndex(indName,tbl.getChunkPtr());
                      return ret ;
                  }
             }
          }
    }// else
    return BackEnd::OK;
}
BackEnd::DbRetVal DatabaseManagerImpl::insertIndexNodeToTable(
                                                       void * chunk   ,
                                                       Table& tbl     ,
                                                       void *indexPtr ,
                                                       IndexInfo *info )
{
    BackEnd::DbRetVal rv =  BackEnd::OK ;
    void *recPtr = NULL;

    /******************** go thru the chunk *******************************/
    ChunkIterator chIter = ((Chunk *)chunk)->getIterator();
    while ((recPtr = chIter.nextElement()) != NULL) {
        // We get the hash index static ptr
        // If it is NULL, should be panic ...
        Index* _hashinxptr = Index::getIndex(hashIndex);
        assert(_hashinxptr);
        rv =   _hashinxptr->insert(&tbl,*tbl.getTrans(),
                                 indexPtr,info,
                                 recPtr,true/*undo flag*/);
        if (rv != BackEnd::OK) {
            break;
        }
    }

    return rv;
}
/**
 * SYNOPSIS :initHashBuckets
 *
 * @param buck
 * @param bucketSize
 */
void DatabaseManagerImpl::initHashBuckets(Bucket *buck, int bucketSize)
{
    memset((char*)buck,0,bucketSize * sizeof(Bucket));
    for (int i=0; i < bucketSize ; i++)
    {
       buck[i].rwlock = PTHREAD_RWLOCK_INITIALIZER;
       buck[i].element_chunk = createUserIndexChunk(sizeof(BUCKET_TREE_ELEMENT));
       BKT_TREE_ARG* thearg = new BKT_TREE_ARG;
       thearg->dbptr= this->db_;
       thearg->chunkptr=buck[i].element_chunk ;
       thearg->alloc_type = allocate_bychunk;
       buck[i].subtree.init_tree((unsigned)0,(int)sizeof(BUCKET_TREE_ELEMENT),
                                          &Index::keys_compare,(void*)thearg);
    }
    return;
}
/**
 * SYNOPSIS :destroyHashBuckets
 *
 * @param buck   Bucket*
 * @param bucketSize
 */
void DatabaseManagerImpl::destroyHashBuckets(Bucket *buck, int bucketSize)
{
     for (int i=0; i < bucketSize ; i++)
    {
      buck[i].subtree.delete_tree();
      deleteUserIndexChunk(buck[i].element_chunk);
      pthread_rwlock_destroy(&buck[i].rwlock);
    }
    return;
}

/**
 *  SYNPOSIS: dropIndex
 *  @parameter  idxname   the nameof index
 *  @parameter  tptr      void*
 *  @parameter  takeLock  the default value is true.
 */
DbRetVal DatabaseManagerImpl::dropIndex(const char *idxname,
                                        void* tptr, bool takeLock)
{
  return dropIndexInt(idxname, tptr, takeLock);
}
/**
 * SYNOPSIS: dropIndexInt
 *
 * @param idxname
 * @param tblptr
 * @param takeLock
 * @return
 */
DbRetVal DatabaseManagerImpl::dropIndexInt(const char *idxname,
                                           void*  tblptr ,
                                           bool takeLock)
{
    DbRetVal rv  = OK;
    void *chunk  = 0;
    void *hchunk = 0;
    void *tptr   = 0;
    Session* _session = current_session ;
    if(!idxname) return rv ;  /*ignore it*/
    if (takeLock) {
        rv = systemDatabase_->getDatabaseMutex();
        if (OK != rv)
      {
        push_warning(_session, COLLAPSAR_ERROR::WARN_LEVEL_WARN,
                     ER_CANT_GET_DB_LOCK, ER(ER_CANT_GET_DB_LOCK));
        return BackEnd::ER_SysInternal;
      }
    }

    // 1 ) remove the entry in CINDEX
    CatalogTableINDEX cIndex(systemDatabase_);
    rv = cIndex.remove(idxname, tblptr, chunk, hchunk, tptr);
    if (OK != rv)
    {
        if (takeLock) systemDatabase_->releaseDatabaseMutex();
        return BackEnd::ER_SysInternal;
    }
    errmsg_printf(ERRMSG_LVL_DBUG,_("[%s] : line [%d] Removing from CINDEX.Index name:[%s].\n"),
                                    __FILE__,__LINE__,idxname);

    // 2) delete the index chunk
    CINDEX *iptr = (CINDEX*)tptr;
      if (iptr->indexType_ == hashIndex)
    {
      ChunkIterator citer = ((Chunk*)chunk)->getIterator();
      Bucket* buckets = (Bucket*)citer.nextElement();
      destroyHashBuckets(buckets,iptr->noOfBuckets_);
    }
    rv = deleteUserChunk((Chunk*)chunk);
    if (OK != rv)
    {
        if (takeLock) systemDatabase_->releaseDatabaseMutex();
        return BackEnd::ER_SysInternal;
    }
    //delete the index hash node chunk
    if (iptr->indexType_ == hashIndex) {
        rv = deleteUserChunk((Chunk*)hchunk);
        if (OK != rv)
        {
          if (takeLock) systemDatabase_->releaseDatabaseMutex();
          return BackEnd::ER_SysInternal;
        }
    }

    Chunk *chunkNode =systemDatabase_->getSystemDatabaseChunk(UserChunkTableId);
    chunkNode->free(systemDatabase_, (Chunk *) chunk);
    if (iptr->indexType_ == hashIndex) {
        chunkNode->free(systemDatabase_, (Chunk *) hchunk);
    }
    if (takeLock) systemDatabase_->releaseDatabaseMutex();
    //Take table lock
    errmsg_printf(ERRMSG_LVL_DBUG, _("[%s]: line [%d] Dropped index:[%s] successfully.\n"),
                  __FILE__,__LINE__,idxname);
    return OK;
}
BackEnd::DbRetVal DatabaseManagerImpl::printIndexInfo(char *name)
{
    CatalogTableINDEX cIndex(systemDatabase_);
    DbRetVal rv = OK;
    void *chunk = NULL, *hchunk = NULL;
    void *tptr =NULL;
    rv = cIndex.get(name, chunk, hchunk, tptr);
    if (OK != rv) return rv;
    printf("<IndexName> %s </IndexName>\n", name);
    printf("<Unique> %d </Unique>\n", CatalogTableINDEX::getUnique(tptr));
    Chunk *ch = (Chunk*) chunk;
    printf("<HashBucket>\n");
    printf("  <TotalPages> %ld </TotalPages>\n", (long)ch->totalPages());
    printf("  <TotalBuckets> %d </TotalBuckets> \n",
           CatalogTableINDEX::getNoOfBuckets(tptr));
    printf("</HashBucket>\n");

    ch = (Chunk*) hchunk;
    printf("<IndexNodes>\n");
    printf("  <TotalPages> %ld </TotalPages>\n", (long)ch->totalPages());
    printf("  <TotalNodes> %d </TotalNodes>\n",(int) ch->getTotalDataNodes());
    printf("<IndexNodes>\n");
    return BackEnd::OK;
}

BackEnd::DbRetVal DatabaseManagerImpl::registerThread()
{
    BackEnd::DbRetVal rv = BackEnd::OK;
    if (pMgr_ != NULL)
    {
     return BackEnd::ER_Already;
    }
    pMgr_ = new ProcessManager();
    rv = pMgr_->registerThread();
    if (rv ==BackEnd::OK) {
        procSlot = pMgr_->getProcSlot();
    }
    return rv;
}

BackEnd::DbRetVal DatabaseManagerImpl::deregisterThread()
{
    BackEnd::DbRetVal rv = BackEnd::OK;
    if (pMgr_ != NULL)
    {
        rv = pMgr_->deregisterThread(procSlot);
        delete pMgr_;
        pMgr_ = NULL;
    }
    return rv;
}

bool DatabaseManagerImpl::isAnyOneRegistered()
{
    if (pMgr_ != NULL) return pMgr_->isAnyOneRegistered();
    return true;
}

/**
 * SYNOPSIS newRecord
 *
 *
 * @param tbl             the table reference
 * @param record          const unsigned char*  the record buff
 * @return   0 on OK
 */
BackEnd::DbRetVal DatabaseManagerImpl::newRecord(Table& tbl,
                                       const unsigned char* record)
{
  DbRetVal ret =OK;
  errkey   = -1   ;
  void *newtptr = ((Chunk*)tbl.getChunkPtr())->allocate(db_, &ret);
  if (!newtptr)
  {
    db_error(ER_OUTOFMEMORY, MYF(0), tbl.s->reclength);
    return(BackEnd::DbRetVal) ER_OUTOFMEMORY ;
  }
  ret = lMgr_->getExclusiveLock(newtptr, tbl.getTrans());
   if (OK != ret)
  {
   ((Chunk*)tbl.getChunkPtr())->free(db_, newtptr);
     return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;
  }
  process_record_data_to_chunkset(static_cast<CTABLE*>(tbl.getCTABLEPtr()),
                                                       record,
                                                 (unsigned char* )newtptr,0);
  BackEnd::ListIterator liter = tbl.idxInfoList.getIterator();
  CatalogTableINDEX cIndex(systemDatabase_) ;
  int key_position  = 0;
  while (liter.hasElement()){
    Index* _hashinxptr = Index::getIndex(hashIndex);
    assert(_hashinxptr);

    IndexInfo* info =(IndexInfo*) liter.nextElement();
    CINDEX* indexPtr = (CINDEX*)((HashIndexInfo*)info)->indexPtr;
    if(!indexPtr)
      indexPtr=cIndex.getIndexPtrByName(tbl.getCTABLEPtr(),info->idxName);
    Transaction** trans = tbl.getTrans();
    Transaction* thetrans = trans? *trans:NULL;
    ret =   _hashinxptr->insert(&tbl,thetrans,indexPtr,info,
                            newtptr,true/*undo flag*/);
    if (ret != BackEnd::OK) {
       errkey  = key_position ;
      ((Chunk*)tbl.getChunkPtr())->free(db_, newtptr);
      lMgr_->releaseLock(newtptr);
      return ret ;
    }
    key_position++;
  }
  lMgr_->releaseLock(newtptr);
  mm_handler* mhlder = (mm_handler*)tbl.file;
  mhlder->file->current_ptr=(unsigned char*)newtptr;
  mhlder->file->update|=HA_STATE_AKTIV;

  CTABLE *ctable=(CTABLE*)tbl.getCTABLEPtr();
  {
    SimpleMutex _mutex(ctable->auto_inc_lock);
    if(ctable->auto_key){
      uint64_t value = 0;            /* Store unsigned values here */
      int64_t s_value= 0;            /* Store signed values here */
      const unsigned char *key=  (unsigned char*) record + ctable->auto_key_offset;
      switch (ctable->auto_key_type) {
      case HA_KEYTYPE_INT8:
        s_value= (int64_t) *(char*)key;
      break;
      case HA_KEYTYPE_BINARY:
        value=(uint64_t)  *(unsigned char*) key;
      break;
      case HA_KEYTYPE_SHORT_INT:
        s_value= (int64_t) sint2korr(key);
      break;
      case HA_KEYTYPE_USHORT_INT:
        value=(uint64_t) uint2korr(key);
      break;
      case HA_KEYTYPE_LONG_INT:
        s_value= (int64_t) sint4korr(key);
      break;
      case HA_KEYTYPE_ULONG_INT:
        value=(uint64_t) uint4korr(key);
      break;
      case HA_KEYTYPE_INT24:
        s_value= (int64_t) sint3korr(key);
      break;
      case HA_KEYTYPE_UINT24:
        value=(uint64_t) uint3korr(key);
      break;
      case HA_KEYTYPE_FLOAT:                        /* This shouldn't be used */
      {
        float f_1;
        float4get(f_1,key);
        /* Ignore negative values */
        value = (f_1 < (float) 0.0) ? 0 : (uint64_t) f_1;
        break;
      }
      case HA_KEYTYPE_DOUBLE:                       /* This shouldn't be used */
      {
        double f_1;
        float8get(f_1,key);
        /* Ignore negative values */
        value = (f_1 < 0.0) ? 0 : (uint64_t) f_1;
        break;
      }
      case HA_KEYTYPE_LONGLONG:
        s_value= sint8korr(key);
      break;
      case HA_KEYTYPE_ULONGLONG:
        value= uint8korr(key);
      break;
      default:
      assert(0);
      value=0;      /* Error */
      break;
    }
    set_if_bigger(ctable->auto_increment,
                    (s_value > 0) ? (uint64_t) s_value : value);
   }
  }
  return OK ;
}

/**
 * SYNOPSIS: delete_all_rows
 *
 * @param   table
 * @return  OK on success
 */

BackEnd::DbRetVal DatabaseManagerImpl::delete_all_rows(Table& table)
{
  BackEnd::DbRetVal rv =  BackEnd::OK ;
  void *recPtr = NULL;
  CatalogTableINDEX cIndex(systemDatabase_) ;
  void *tblptr = table.getCTABLEPtr();
  Chunk* chunk =(Chunk*) table.getChunkPtr();
  ChunkIterator chIter = ((Chunk *)chunk)->getIterator();
  Index* _hashinxptr = Index::getIndex(hashIndex);
  Transaction** trans = table.getTrans();
  Transaction* thetrans = trans? *trans:NULL;
  mm_handler* mhlder = (mm_handler*)table.file;
  while ((recPtr = chIter.nextElement()) != NULL) {
    rv = systemDatabase_->getDatabaseMutex();
    if (OK != rv)return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;

    rv = lMgr_->getExclusiveLock(recPtr, table.getTrans());
    if (OK != rv)
    {
        systemDatabase_->releaseDatabaseMutex();
        return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;
    }
    BackEnd::ListIterator iter = table.idxInfoList.getIterator();
    while( iter.hasElement()){
    IndexInfo* info = static_cast<IndexInfo *>(iter.nextElement());
    CINDEX* indexPtr = (CINDEX*)((HashIndexInfo*)info)->indexPtr;
    if(!indexPtr){
      indexPtr = cIndex.getIndexPtrByName(tblptr,info->idxName);
      info->indexPtr= (unsigned char*)indexPtr;
    }
    rv =   _hashinxptr->remove(&table,thetrans,indexPtr,info,
                                  recPtr,recPtr,true/*undo flag*/);
     if(OK != rv)
    {
      lMgr_->releaseLock(recPtr);
      systemDatabase_->releaseDatabaseMutex();
      goto DELETE_ALL_ROW;
    }

   }
   ((Chunk*)table.getChunkPtr())->free(db_, recPtr);
   lMgr_->releaseLock(recPtr);
   systemDatabase_->releaseDatabaseMutex();
  }

  mhlder->file->hlderinfo.records  = 0;
DELETE_ALL_ROW:
  return rv;
}

/*
 * SYNOPSIS   delete_row
 *
 * @param  table
 * @param  mminfo
 * @param  record
 *
 * @return    0         SUCCESS
 *            else      error
 */
BackEnd::DbRetVal DatabaseManagerImpl::delete_row(Table& table,
                                                  MM_INFO* mminfo,
                                                  unsigned char *record)
{
  BackEnd::DbRetVal rv =  BackEnd::OK ;
  CatalogTableINDEX cIndex(systemDatabase_) ;
  void *tblptr = table.getCTABLEPtr();
  Chunk* chunk =(Chunk*) table.getChunkPtr();
  ChunkIterator chIter = ((Chunk *)chunk)->getIterator();
  Index* _hashinxptr = Index::getIndex(hashIndex);
  Transaction** trans = table.getTrans();
  Transaction* thetrans = trans? *trans:NULL;

  rv = systemDatabase_->getDatabaseMutex();
  if (OK != rv)return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;

  rv = lMgr_->getExclusiveLock(mminfo->current_ptr, table.getTrans());
  if (OK != rv)
  {
     systemDatabase_->releaseDatabaseMutex();
     return(DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;
  }
  BackEnd::ListIterator iter = table.idxInfoList.getIterator();
  while( iter.hasElement()){
    IndexInfo* info = static_cast<IndexInfo *>(iter.nextElement());
    CINDEX* indexPtr = (CINDEX*)((HashIndexInfo*)info)->indexPtr;
    if(!indexPtr)
    {
      indexPtr=cIndex.getIndexPtrByName(tblptr,info->idxName);
      info->indexPtr =(unsigned char*)indexPtr;
    }

    rv =_hashinxptr->remove(&table,thetrans,indexPtr,info,
                            record,mminfo->current_ptr,true/*undo flag*/);
    if(OK != rv)
   {
     lMgr_->releaseLock(mminfo->current_ptr);
     goto DELETE_ERROR;
   }
  }
  ((Chunk*)table.getChunkPtr())->free(db_, mminfo->current_ptr);
  lMgr_->releaseLock(mminfo->current_ptr);
DELETE_ERROR:
  systemDatabase_->releaseDatabaseMutex();
  return (rv) ;
}


void DatabaseManagerImpl::getInfo(MM_INFO* mminfo)
{
     if(!mminfo || !mminfo->table_alias) return ;

     void *tptr  =NULL;
     void *chunk = NULL;
     DbRetVal rv = systemDatabase_->getDatabaseMutex();
     if (OK != rv)
       return ;

     //check whether table already exists
     CatalogTableTABLE cTable(systemDatabase_);
     cTable.getChunkAndTblPtr(mminfo->table_alias, chunk, tptr);
      if (tptr)
     {
        CTABLE* ctable = (CTABLE*)tptr;
        SimpleMutex _mutex(ctable->auto_inc_lock);
        mminfo->auto_increment = ctable->auto_increment +1 ;
        mminfo->auto_key       = ctable->auto_key ;
        mminfo->auto_key_type  = ctable->auto_key_type;

        mminfo->hlderinfo.records =((Chunk*)chunk)->getTotalDataNodes();
        mminfo->hlderinfo.deleted = 0;
        mminfo->hlderinfo.data_length =  (uint64_t)~0;
        mminfo->hlderinfo.index_length = (uint64_t)~0;
     }
     systemDatabase_->releaseDatabaseMutex();
     return ;
}

int DatabaseManagerImpl::mm_rkey(Table& tbl,MM_INFO* mminfo,
                                 unsigned char * record,
                                 int inx,
                                 const unsigned char *key,
                                 key_part_map keypart_map,
                                 enum ha_rkey_function find_flag)
{
  if(inx >= (int)tbl.idxInfoList.size())
     return (my_errno= HA_ERR_WRONG_INDEX);
  IndexInfo* idxinfo = static_cast<IndexInfo*> (tbl.idxInfoList.get(inx+1));
  if (!idxinfo)
     return (my_errno= HA_ERR_WRONG_INDEX);
  CatalogTableINDEX cIndex(systemDatabase_) ;
  void *tblptr = tbl.getCTABLEPtr();
  CTABLE* ctable = (CTABLE*)tblptr;
  CINDEX* indexPtr = (CINDEX*)((HashIndexInfo*)idxinfo)->indexPtr;
  if(!indexPtr)
    indexPtr=cIndex.getIndexPtrByName(tblptr,idxinfo->idxName);
  ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);
  Bucket* buckets = (Bucket*)citer.nextElement();

  mminfo->lastinx = inx ;
  BKT_TREE_ARG custom_arg;
  custom_arg.search_flag= SEARCH_FIND | SEARCH_SAME;
  custom_arg.info  = (IndexInfo*)idxinfo;
  custom_arg.key_length= HashIndex::rb_pack_key(idxinfo, (unsigned char*) mminfo->lastkey,
                                       (unsigned char*) key, keypart_map);
  custom_arg.search_type = get_element;
  /* for next rkey() after deletion */
  if (find_flag == HA_READ_AFTER_KEY)
    mminfo->last_find_flag= HA_READ_KEY_OR_NEXT;
  else if (find_flag == HA_READ_BEFORE_KEY)
    mminfo->last_find_flag= HA_READ_KEY_OR_PREV;
  else
    mminfo->last_find_flag= find_flag;
  uint32_t bucketNo =(uint32_t)(HashIndex::cal_rec_key_hash((unsigned char*)key,
                                custom_arg.key_length)%((HashIndexInfo*)idxinfo)->noOfBuckets);

  Bucket *bucket = &buckets[bucketNo];

  BUCKET_TREE_ELEMENT* the_element =(BUCKET_TREE_ELEMENT*) bucket->subtree.tree_search_key
                                                                 (mminfo->lastkey,
                                                                  mminfo->parents,
                                                                  &mminfo->last_pos,
                                                                  find_flag, &custom_arg);
  if(!the_element) {
    mminfo->update= 0;
    mminfo->current_ptr = 0 ;
    return(my_errno= HA_ERR_KEY_NOT_FOUND);
  }
  HashIndexNode* thenode =0;
  if(HA_READ_PREFIX_LAST == find_flag)
  {
     thenode = (HashIndexNode*)the_element->bucketList_->getBucketListTail();
  }else
     thenode = (HashIndexNode*)the_element->bucketList_->getBucketListHead();

  if (!thenode){
    mminfo->update= 0;
    mminfo->current_ptr = 0 ;
    return(my_errno= HA_ERR_KEY_NOT_FOUND);
  }
  mminfo->current_hIdxNode = thenode;
  mminfo->lastkey_len = custom_arg.key_length ;
  mminfo->current_ptr = (unsigned char*)thenode->ptrToTuple_;
  mm_extract_record(ctable,record,(const unsigned char *)thenode->ptrToTuple_);
  mminfo->update= HA_STATE_AKTIV  ;
  return (0) ;
}

int DatabaseManagerImpl::mm_rfirst(Table& tbl,
                                   MM_INFO* mminfo,
                                   unsigned char* record,
                                   int inx)
{
  if(inx >= (int)tbl.idxInfoList.size())
       return (my_errno= HA_ERR_WRONG_INDEX);
  IndexInfo* idxinfo = static_cast<IndexInfo*> (tbl.idxInfoList.get(inx+1));
  if (!idxinfo)
     return (my_errno= HA_ERR_WRONG_INDEX);
  CatalogTableINDEX cIndex(systemDatabase_) ;
  void *tblptr = tbl.getCTABLEPtr();
  CTABLE* ctable = (CTABLE*)tblptr;
  CINDEX* indexPtr = (CINDEX*)((HashIndexInfo*)idxinfo)->indexPtr;
  if(!indexPtr)
    indexPtr=cIndex.getIndexPtrByName(tblptr,idxinfo->idxName);
  ChunkIterator citer = CatalogTableINDEX::getIterator(indexPtr);
  Bucket* buckets = (Bucket*)citer.nextElement();

  mminfo->lastinx = inx ;
  BKT_TREE_ARG custom_arg;
  custom_arg.search_flag= SEARCH_FIND | SEARCH_SAME;
  custom_arg.info  = (IndexInfo*)idxinfo;
  custom_arg.key_length= mminfo->lastkey_len;
  custom_arg.search_type = get_element;

  uint32_t bucketNo =(uint32_t)(HashIndex::cal_rec_key_hash((unsigned char*)record,
                                custom_arg.key_length)%((HashIndexInfo*)idxinfo)->noOfBuckets);

  Bucket *bucket = &buckets[bucketNo];

  BUCKET_TREE_ELEMENT* the_element =(BUCKET_TREE_ELEMENT*) bucket->subtree.tree_search_key
                                                                 (mminfo->lastkey,
                                                                  mminfo->parents,
                                                                  &mminfo->last_pos,
                                                                  HA_READ_KEY_EXACT, &custom_arg);
  if(!the_element) {
    mminfo->update= 0;
    mminfo->current_ptr = 0 ;
    return(my_errno = HA_ERR_END_OF_FILE);
  }
  HashIndexNode* thenode =0;

  thenode = (HashIndexNode*)the_element->bucketList_->getBucketListHead();

  if (!thenode){
    mminfo->update= 0;
    mminfo->current_ptr = 0 ;
    return(my_errno = HA_ERR_END_OF_FILE);
  }
  mminfo->current_hIdxNode = thenode;
  /*
     If we're performing index_first on a table that was taken from
     table cache, info->lastkey_len is initialized to previous query.
     Thus we set info->lastkey_len to proper value for subsequent
     heap_rnext() calls.
     This is needed for DELETE queries only, otherwise this variable is
     not used.
     Note that the same workaround may be needed for heap_rlast(), but
     for now heap_rlast() is never used for DELETE queries.
   */

  mminfo->lastkey_len = 0 ;
  mminfo->current_ptr = (unsigned char*)thenode->ptrToTuple_;
  mm_extract_record(ctable,record,(const unsigned char *)thenode->ptrToTuple_);
  mminfo->update= HA_STATE_AKTIV  ;
  return (0) ;
}

int DatabaseManagerImpl::mm_rnext (Table& ,MM_INFO* ,
                                   unsigned char *, int )
{


   return (0);
}
int DatabaseManagerImpl::mm_rnext_same(Table& ,//tbl,
                                       MM_INFO*,// mminfo,
                                       unsigned char*)// *buf)
{
  return (0) ;
}
uint32_t DatabaseManagerImpl::get_encoded_data_length(CTABLE* tptr,
                                               const unsigned char* )
{
   uint32_t  dst_offset;
   dst_offset = tptr->fixed_data_length;
   return dst_offset;
}

uint32_t DatabaseManagerImpl::process_record_data_to_chunkset
                                        (CTABLE* tptr,
                                         const unsigned char *record,
                                  unsigned char* pos, uint32_t is_compare)
{
  unsigned char* curr_chunk= pos;
  if (is_compare)
  {
    if (memcmp(curr_chunk, record, (size_t) tptr->fixed_data_length))
    {
      return 1;
    }
  }else
  {
    memcpy(curr_chunk, record, (size_t) tptr->fixed_data_length);
  }
  return (0);
}

void DatabaseManagerImpl::mm_extract_record   (CTABLE* tptr,
                                               unsigned char *record,
                                               const unsigned char *pos)
{
  const unsigned char* curr_chunk= pos;
  memcpy(record, curr_chunk, (size_t) tptr->fixed_data_length);
}
void DatabaseManagerImpl::printUsageStatistics()
{
    pMgr_->printUsageStatistics();
    tMgr_->printUsageStatistics();
    lMgr_->printUsageStatistics();
}

void DatabaseManagerImpl::printDebugLockInfo()
{
    lMgr_->printDebugInfo();
}

void DatabaseManagerImpl::printDebugTransInfo()
{
    tMgr_->printDebugInfo(systemDatabase_);
}
void DatabaseManagerImpl::printDebugProcInfo()
{
    pMgr_->printDebugInfo();
}
void DatabaseManagerImpl::printDebugChunkInfo()
{
}
ChunkIterator DatabaseManagerImpl::getSystemTableIterator(CatalogTableID id)
{
    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(id);
    return fChunk->getIterator();
}

Chunk* DatabaseManagerImpl::getSystemTableChunk(CatalogTableID id)
{
    return systemDatabase_->getSystemDatabaseChunk(id);
}
